package influxdbtool

import (
	"fmt"
	"github.com/influxdata/line-protocol/v2/lineprotocol"
	"log/slog"
	"time"
)

// LineEntry line-protocol 解析后的格式
type LineEntry struct {
	Measurement string
	Tags        []TagKeyValue
	Fields      []FieldKeyValue
	Time        time.Time
}

func (entry *LineEntry) Tag(key string) string {
	for _, tag := range entry.Tags {
		if tag.Key == key {
			return tag.Value
		}
	}
	return ""
}

func (entry *LineEntry) Field(key string) interface{} {
	for _, field := range entry.Fields {
		if field.Key == key {
			return field.Value
		}
	}
	return nil
}

type TagKeyValue struct {
	Key, Value string
}

type FieldKeyValue struct {
	Key   string
	Value interface{}
}

// Preprocess 对 line-protocol 原始数据进行预处理
func Preprocess(rawDataChan <-chan []byte, entryChan chan<- LineEntry) {
	for input := range rawDataChan {
		decoder := lineprotocol.NewDecoderWithBytes(input)

		for decoder.Next() {
			m, err := decoder.Measurement()
			if err != nil {
				slog.Debug("", "Error decoding measurement", err)
				continue
			}

			tags := make([]TagKeyValue, 0)
			for {
				key, val, err2 := decoder.NextTag()
				if err2 != nil {
					slog.Debug("", "Error decoding tag", err2)
					break
				}
				if key == nil {
					break
				}

				tags = append(tags, TagKeyValue{string(key), string(val)})
			}

			fields := make([]FieldKeyValue, 0)
			for {
				key, val, err2 := decoder.NextField()
				if err2 != nil {
					slog.Debug("", "Error decoding field", err2)
					break
				}
				if key == nil {
					break
				}

				fields = append(fields, FieldKeyValue{string(key), val.Interface()})
			}

			t, err := decoder.Time(lineprotocol.Nanosecond, time.Time{})
			if err != nil {
				fmt.Printf("Error decoding time: %s\n", err)
			}

			a := LineEntry{
				Measurement: string(m),
				Tags:        tags,
				Fields:      fields,
				Time:        t,
			}
			entryChan <- a
		}
	}
}
